// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_CLIENT_ENVIRONMENT_HPP_
#define _XBASE_CLIENT_ENVIRONMENT_HPP_

#include <list>

#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>
#include <boost/asio/deadline_timer.hpp>

#include "scanner.hpp"
#include "row_mutator.hpp"
#include "table_mutator.hpp"
#include "location_cache.hpp"
#include "session_manager.hpp"
#include "root_meta_cache.hpp"
#include "io_service_pool.hpp"
#include "parallel_scanner.hpp"
#include "table_info_cache.hpp"
#include "multi_row_mutator.hpp"

namespace xbase
{

	class client_environment
		: private boost::noncopyable
	{
	public:
		typedef client_environment this_type;
		typedef boost::shared_ptr<this_type> ptr_type;

		typedef session_manager::ptr_type session_manager_ptr;
		typedef session_manager::endpoint_type endpoint_type;

		typedef root_meta_cache::ptr_type root_meta_cache_ptr;

		typedef table_info_cache::ptr_type table_info_cache_ptr;

		typedef location_cache::ptr_type location_cache_ptr;

		typedef std::list<row_mutator_ptr> row_mutator_list;
		typedef std::list<multi_row_mutator_ptr> multi_row_mutator_list;
		typedef std::list<scanner_ptr> scanner_list;
		typedef std::list<parallel_scanner_ptr> parallel_scanner_list;

		typedef boost::asio::deadline_timer timer_type;
		typedef boost::shared_ptr<timer_type> timer_ptr;

	public:
		client_environment(client_options_ptr options,
				   io_service_pool_ptr pool,
				   const std::string &master_ip,
				   unsigned short master_port);

		~client_environment();

		io_service_pool& get_io_service_pool()
		{
			return *m_io_service_pool;
		}

		const endpoint_type& get_master_address() const
		{
			return m_master_address;
		}

		session_manager& get_session_manager()
		{
			return *m_session_manager;
		}

		root_meta_cache& get_root_meta_cache()
		{
			return *m_root_meta_cache;
		}

		table_info_cache& get_table_info_cache()
		{
			return *m_table_info_cache;
		}

		location_cache &get_location_cache()
		{
			return *m_location_cache;
		}

		void async_open_table(const std::string &table_name, table_info_cache::handler_type handler);

		row_mutator_ptr new_row_mutator();

		multi_row_mutator_ptr new_multi_row_mutator();

		parallel_scanner_ptr new_parallel_scanner(const scan_info &si, parallel_scanner::handler_type handler);

		scanner_ptr new_scanner(const scan_info &si);

		meta_tablet_scanner_ptr new_meta_tablet_scanner(std::string &hostname, uint16_t port,
			std::string &start_row, std::string &end_row);

		void async_apply(const row_mutation &mutation, row_mutator::handler_type handler);

		void async_apply(const row_mutation &mutation, uint64_t timestamp, row_mutator::handler_type handler);

		void async_apply(const std::string &table_name, const multi_row_mutator::container_ptr &mutations,
			uint64_t timestamp, multi_row_mutator::handler_type handler);

		void async_apply(const std::string &table_name, const multi_row_mutator::container_ptr &mutations,
			multi_row_mutator::handler_type handler);

		void async_apply(const table_creation &creation, table_mutator::create_handler_type handler);

		void async_apply(const table_deletion &deletion, table_mutator::delete_handler_type handler);

		void async_apply(const table_lister &lister, table_mutator::list_handler_type handler);

	private:
		this_type *this_ptr()
		{
			return this;
		}

		void on_stat_dump(const boost::system::error_code& error);

		void on_new_scanner(scanner_ptr scanner_p);

	private:
		client_options_ptr m_options;
		io_service_pool_ptr m_io_service_pool;
		endpoint_type m_master_address;

		// others: retry_mode ...

		session_manager_ptr m_session_manager;
		// caches
		root_meta_cache_ptr m_root_meta_cache;
		table_info_cache_ptr m_table_info_cache;
		location_cache_ptr m_location_cache;

		timer_ptr m_stat_dump_timer;

		row_mutator_list m_row_mutator_list;
		multi_row_mutator_list m_multi_row_mutator_list;
		scanner_list m_scanner_list;
		parallel_scanner_list m_parrallel_scanner_list;

		stat m_stat;
	};

	typedef client_environment::ptr_type client_environment_ptr;

	class row_mutator_pool
		: private boost::noncopyable
	{
	public:
		row_mutator_pool(client_environment &env,
				 std::size_t max_size = 10000)
			: m_env(env)
			, m_max_size(max_size) {}

		row_mutator_ptr new_row_mutator() {
			boost::mutex::scoped_lock lock(m_mutex);
			if(m_free_mutator.empty())
			{
				return m_env.new_row_mutator();
			}
			else
			{
				row_mutator_ptr ptr = m_free_mutator.back();
				m_free_mutator.pop_back();
				return ptr;
			}
		}

		void delete_row_mutator(row_mutator_ptr ptr) {
			boost::mutex::scoped_lock lock(m_mutex);
			if(m_free_mutator.size() < m_max_size)
				m_free_mutator.push_front(ptr);
		}

	private:
		client_environment &m_env;
		boost::mutex m_mutex;
		std::deque<row_mutator_ptr> m_free_mutator;
		std::size_t m_max_size;
	};


} // namespace xbase

#endif	// _XBASE_CLIENT_ENVIRONMENT_HPP_
